package indi.zhaosheng.rocktmq.producer;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author muluo
 * @description
 * @date 2020/5/5 15:46
 */
public abstract class BaseOrderlyProducer extends BaseProducer {

    private static final Logger logger = LoggerFactory.getLogger(BaseOrderlyProducer.class);

    protected BaseOrderlyProducer(String topic, String producerGroup) {
        super(topic, producerGroup);
    }

    private MessageQueueSelector messageQueueSelector = (mqs, msg, arg) -> {
        Integer id = (Integer) arg;
        int index = id % mqs.size();
        return mqs.get(index);
    };

    public boolean sendOrderlyMsg(String tag, Object msg, int orderId) {
        Message message = new Message(topic, tag, JSON.toJSONBytes(msg));
        try {
            SendResult sendResult = defaultMQProducer.send(message, messageQueueSelector, orderId);
            logger.info("msg send to topic:{}, msgId:{}, status:{}",
                    this.topic, sendResult.getMsgId(), sendResult.getSendStatus());
            return true;
        } catch (Exception e) {
            logger.error("msg send to topic:{}, body:{}, error context:{}",
                    this.topic, JSON.toJSONString(message), e);
        }
        return false;
    }


}
